package com.crazymaker.demo.hystrix;

import com.crazymaker.springcloud.standard.utils.HttpRequestUtil;
import com.netflix.hystrix.*;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import rx.Observer;

import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

@Slf4j
public class CommandTester
{

    /**
     * 根据实际的地址调整
     */
    private static final String TEST_URL =
            "http://192.168.233.128:8092/api/demo/hello/v1";

    /**
     * 错误的url
     */
    private static final String ERROR_URL = "www.baidu.xinlang.com";

    /**
     * 获取配置
     *
     * @param groupKey
     * @param commandKey
     * @param threadPoolKey
     * @return
     */
    private static HystrixCommand.Setter getSetter(
            String groupKey,
            String commandKey,
            String threadPoolKey)
    {
        /**
         * 命令参数配置
         */
        HystrixCommandProperties.Setter commandPropertiesSetter =
                HystrixCommandProperties.Setter()
                        //至少有3个请求,熔断器才进行错误率的计算
                        .withCircuitBreakerRequestVolumeThreshold(3)
                        //熔断器中断请求5秒后会进入半打开状态,放部分流量过去重试
                        .withCircuitBreakerSleepWindowInMilliseconds(5000)
                        //错误率超过60%,快速失败
                        .withCircuitBreakerErrorThresholdPercentage(60)
                        //启用超时
                        .withExecutionTimeoutEnabled(true)
                        //执行的超时时间，默认为 1000 ms
                        .withExecutionTimeoutInMilliseconds(5000)
                        // 可统计的滚动窗口内的buckets数量,用于熔断器和指标发布
                        .withMetricsRollingStatisticalWindowBuckets(10)
                        // 可统计的滚动窗口的时间长度
                        // 这段时间内的执行数据用于熔断器和指标发布
                        .withMetricsRollingStatisticalWindowInMilliseconds(10000);
        /**
         * 线程池配置
         */
        HystrixThreadPoolProperties.Setter threadPoolPropertiesSetter =
                HystrixThreadPoolProperties.Setter()
                        //这里我们设置了线程池大小为10
                        .withCoreSize(10)
                        .withMaximumSize(10);
        /**
         * 命令配置
         */
        HystrixCommand.Setter commandSetter = HystrixCommand.Setter
                .withGroupKey(HystrixCommandGroupKey.Factory.asKey(groupKey))
                .andCommandKey(
                        HystrixCommandKey.Factory.asKey(commandKey))
                .andThreadPoolKey(
                        HystrixThreadPoolKey.Factory.asKey(threadPoolKey))
                .andCommandPropertiesDefaults(
                        commandPropertiesSetter)
                .andThreadPoolPropertiesDefaults(
                        threadPoolPropertiesSetter);
        return commandSetter;
    }

    /**
     * 测试url 是否可达
     *
     * @throws Exception
     */
    @Test
    public void testUrl()
    {
        String responseData = HttpRequestUtil.simpleGet(TEST_URL);
        log.info(responseData);
    }


    @Test
    public void testExecute() throws Exception
    {
        /**
         * 使用统一配置
         */
        HystrixCommand.Setter setter = getSetter(
                "group-1",
                "testCommand",
                "testThreadPool" );

        for (int i = 1; i < 10; i++)
        {
            String result =
                    new HttpGetterCommand(TEST_URL, setter).execute();
            log.info("第{}次请求的结果：{}", i, result);
        }
        Thread.sleep(Integer.MAX_VALUE);

    }


    @Test
    public void testQueue() throws Exception
    {
        /**
         * 使用统一配置
         */
        HystrixCommand.Setter setter = getSetter(
                "group-1",
                "testCommand",
                "testThreadPool" );
        List<Future<String>> flist = new LinkedList<>();

        /**
         * 同时发起5个异步的请求
         */
        for (int i = 0; i < 5; i++)
        {
            Future<String> future =
                    new HttpGetterCommand(TEST_URL, setter).queue();

            flist.add(future);

        }
        /**
         * 阻塞处理异步请求的结果
         */
        Iterator<Future<String>> it = flist.iterator();
        int count = 1;
        while (it.hasNext())
        {
            Future<String> future = it.next();
            String result = future.get(10, TimeUnit.SECONDS);
            log.info("第{}次请求的结果：{}", count++, result);
        }
        Thread.sleep(Integer.MAX_VALUE);
    }

    @Test
    public void testObserve() throws Exception
    {
        /**
         * 使用统一配置
         */
        HystrixCommand.Setter setter = getSetter(
                "group-1",
                "testCommand",
                "testThreadPool" );

        for (int i = 0; i < 50; i++)
        {
            Thread.sleep(2);

            int finalI = i;
            new HttpGetterCommand(TEST_URL, setter)
                    .observe()
                    .subscribe(new Observer<String>()
                    {
                        @Override
                        public void onCompleted()
                        {
                            log.info("第{}次请求结束", finalI);
                        }

                        @Override
                        public void onError(Throwable e)
                        {
                            log.error("第{}次请求 failed:{}", finalI, e.getMessage());
                        }

                        @Override
                        public void onNext(String result)
                        {
                            log.info("第{}次请求的结果：{}", finalI, result);
                        }
                    });

        }
        Thread.sleep(Integer.MAX_VALUE);
    }

    @Test
    public void testObserveLamda() throws Exception
    {
        /**
         * 使用统一配置
         */
        HystrixCommand.Setter setter = getSetter(
                "group-1",
                "testCommand",
                "testThreadPool" );

        for (int i = 0; i < 50; i++)
        {
            Thread.sleep(2);

            new HttpGetterCommand(TEST_URL, setter)
                    .observe()
                    .subscribe(s -> log.info(" onNext ::" + s));
        }
        Thread.sleep(Integer.MAX_VALUE);

    }

    @Test
    public void testObserveNoSubscribe() throws Exception
    {
        /**
         * 使用统一配置
         */
        HystrixCommand.Setter setter = getSetter(
                "group-1",
                "testCommand",
                "testThreadPool" );

        for (int i = 0; i < 50; i++)
        {
            Thread.sleep(2);
            new HttpGetterCommand(TEST_URL, setter)
                    .observe();
        }
        Thread.sleep(Integer.MAX_VALUE);

    }

    @Test
    public void testToObservable() throws Exception
    {
        /**
         * 使用统一配置
         */
        HystrixCommand.Setter setter = getSetter(
                "group-1",
                "testCommand",
                "testThreadPool" );

        for (int i = 0; i < 50; i++)
        {
            Thread.sleep(2);

            new HttpGetterCommand(TEST_URL, setter)
                    .toObservable()
                    .subscribe(s -> log.info(" onNext ::" + s));
        }
        Thread.sleep(Integer.MAX_VALUE);
    }

    /**
     * 测试：线程池满了
     */

    @Test
    public void testThreadPool() throws Exception
    {

        /**
         * 使用统一配置
         */
        HystrixCommand.Setter setter = getSetter(
                "group-1",
                "testCommand",
                "testThreadPool" );

        for (int i = 1; i <= 46; i++)
        {
            if (i % 15 == 0)
            {
                Thread.sleep(2);
            }
            new HttpGetterCommand(TEST_URL, setter)
                    .toObservable()
                    .subscribe(s -> log.info(" onNext ::" + s));
        }

        Thread.sleep(Integer.MAX_VALUE);

    }

    /**
     * 测试:失败率导致快速失败
     */

    @Test
    public void testFastFallback() throws Exception
    {

        /**
         * 使用统一配置
         */
        HystrixCommand.Setter setter = getSetter(
                "group-1",
                "testCommand",
                "testThreadPool" );

        for (int i = 1; i <= 46; i++)
        {
            if (i % 15 == 1)
            {
                Thread.sleep(2);

                for (int j = 1; j <= 46; j++)
                {

                    new HttpGetterCommand(ERROR_URL, setter)
                            .toObservable()
                            .subscribe(s -> log.info(" onNext ::" + s));
                }

            }

            new HttpGetterCommand(TEST_URL, setter)
                    .toObservable()
                    .subscribe(s -> log.info(" onNext ::" + s));
        }

        Thread.sleep(Integer.MAX_VALUE);

    }
}